package com.atguigu.flink.sql.time;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * Created by 黄凯 on 2023/6/26 0026 18:16
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 *
 * 在SQL中声明时间字段
 */
public class Flink02_SqlTime {

    public static void main(String[] args) {

        // 创建表环境
        TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());

        //连接器表
        //2. File Connector read
        String readSql=
                " create table t1 (" +
                        " `user` STRING , " +
                        " url STRING , " +
                        " ts BIGINT, " +
                        //处理时间
                        " pt AS PROCTIME() ," +
                        //事件时间
                        " et AS TO_TIMESTAMP_LTZ(ts, 3) ," +
                        " watermark for et as et - interval '2' second " +
                        ") WITH ( " +
                        " 'connector' = 'filesystem' , " +
                        " 'path' = 'input/clicks.txt', " +
                        " 'format' = 'csv' " +
                        ")" ;

        tableEnv.executeSql(readSql);

        //将Table对象注册为环境中的表
        //tableEnv.createTemporaryView("t1" , table);

        //将环境中的表 转换成 Table
        Table table = tableEnv.from("t1");
        table.printSchema();

    }

}
